﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Runtime.Serialization.Formatters.Binary;
using System.Security.Cryptography;
using PhiStore.ObjectManaging;
using PhiStore.Packages;
using PhiStore.Packages.InternalPackages;
using PhiStore.Policies;
using PhiStore.Security;
using PhiStore.Threading;

namespace PhiStore.Communication
{
    internal partial class Worker : Boostable
    {
        private TcpClient connection;
        private BinaryFormatter formatter = new BinaryFormatter();
        RSACryptoServiceProvider rsa = new RSACryptoServiceProvider();

        private LockManager locks = new LockManager();

        private Channel<object> fromObjectManager = new Channel<object>(5);

        private Stream outputStream = null;
        private Stream inputStream = null;

        private ObjectLock objectLock = null;
        private string currentObjectName = "";

        bool verified = false;
        bool serverVerified = false;

        AesCryptoServiceProvider aes = null;

        PeerContainer externalPeer = null;
        private PeerContainer ExternalPeer
        {
            get
            {
                return externalPeer;
            }
            set
            {
                Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "External peer marked: " + value.BC.ToString());
                externalPeer = value;
            }
        }

        public Worker(TcpClient connection)
        {
            this.connection = connection;
            this.connection.ReceiveTimeout = 30000;
            this.connection.SendTimeout = 30000;

            rsa.ImportParameters(EncryptionSettings.Instance.PrivateKey);
        }

        public void Run()
        {
            Log.Instance.AddLogEntry("WORKER", "Worker Started with a connection from " + connection.Client.RemoteEndPoint.ToString());

            try
            {
                outputStream = connection.GetStream();
                inputStream = connection.GetStream();

                while (true)
                {
                    //aborting uncommitted modifications if they are not committed within 30 sec.
                    foreach (string key in locks.GetDeadKeys())
                    {
                        doAbort(key);
                    }

                    //reset
                    currentObjectName = "";
                    objectLock = null;

                    PhiPackage CurrentAction = ReceiveAction();
                    
                    //MANUEL LATENCY
                    //Thread.Sleep(25);

                    PerformAction(CurrentAction);

                }
            }
            catch (IOException)//timeout
            {
                Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "CommunicationWorker Timeout");
                //thread going down!
            }
            catch (Exception e)
            {
                Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), e.ToString());
            }
            finally
            {
                if (outputStream != null)
                    outputStream.Close();
                if (inputStream != null)
                    inputStream.Close();

                this.connection.Close();

                foreach (string key in locks.GetAllLocks()) //aborting all uncommitted
                {
                    Log.Instance.AddLogEntry("CommunicationWorker", "Aborting " + key);
                    try
                    {
                        doAbort(key);
                    }
                    catch
                    {

                    }
                }
            }
            ActivityMonitor.Instance.DecreaseActivity();
        }

        private void PerformAction(PhiPackage CurrentAction)
        {
            //Thread.Sleep(1000);
            Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(),
                "Recieved an package: " + CurrentAction.ToString());
            object result = ReturnCode.Failure;//default result
            if (!verified) //Connection not verified yet
            {
                #region Action which may be performed without verification. E.g. login stuff
                if (CurrentAction is WhoAreYou)
                {
                    result = PhiServer.Instance.BroadcastObject;
                }
                else if (CurrentAction is Connect)
                {
                    result = CredentialManager.Instance.performAction(CurrentAction);

                    verified = (ReturnCode)result == ReturnCode.Success;
                    serverVerified = verified && ((CurrentAction as Connect).Credentials is ServerCredentialContainer);

                    if ((CurrentAction as Connect).EncryptionKey.Length > 0)
                    {
                        aes = new AesCryptoServiceProvider();
                        aes.Key = (CurrentAction as Connect).EncryptionKey;
                        //aes.GenerateIV();
                        aes.IV = (CurrentAction as Connect).IV;
                    }

                }
                else
                {
                    result = ReturnCode.ConnectionNotVerified;
                }

                SendReply(result);
                #endregion
            }
            else //connection is verified
            {
                #region client operations
                if (CurrentAction is Disconnect)
                {
                    throw new IOException("DISCONNECT");
                }
                else if (CurrentAction is GetPolicy)
                {
                    PhiServer.Instance.ObjectManagerChannel.Send(new InternalGetPolicy((CurrentAction as GetPolicy).Key, fromObjectManager));

                    object o = fromObjectManager.Receive();
                    if (o is Policy)
                        SendReply(new ReadReply(ReturnCode.Success, o));
                    else
                        SendReply(new ReadReply((ReturnCode)o, o));

                }
                else if (CurrentAction is Create)
                {
                    Create((Create)CurrentAction);
                }
                else if (CurrentAction is Read)
                {
                    ReadObject((Read)CurrentAction);
                }
                else if (CurrentAction is Update)
                {
                    UpdateObject((Update)CurrentAction);
                }
                else if (CurrentAction is Delete)
                {
                    DeleteObject((Delete)CurrentAction);
                }
                else if (CurrentAction is ListAppend)
                {
                    Append((ListAppend)CurrentAction);
                }
                else if (CurrentAction is ListDelete)
                {
                    DeleteListEntries((ListDelete)CurrentAction);
                }
                else if (CurrentAction is ListCount)
                {
                    GetCount((ListCount)CurrentAction);
                }
                else if (CurrentAction is ListPop)
                {
                    Pop((ListPop)CurrentAction);
                }
                #endregion
                else if (serverVerified)
                {
                    if (CurrentAction is Abort)
                    {
                        Abort((Abort)CurrentAction);
                    }
                    else if (CurrentAction is Commit)
                    {
                        Commit((Commit)CurrentAction);
                    }
                    else if (CurrentAction is NegotiationInit)
                    {
                        Negotiation((NegotiationInit)CurrentAction);
                    }
                    else if (CurrentAction is NegotiationCommit)
                    {
                        NegotiationFinished((NegotiationCommit)CurrentAction);
                    }
                    else if (CurrentAction is AddPeer)
                    {
                        AddPeer((AddPeer)CurrentAction);
                    }
                    else if (CurrentAction is AddRemotePeer)
                    {
                        AddRemotePeer((AddRemotePeer)CurrentAction);
                    }
                    else if (CurrentAction is ListCredentials)
                    {
                        SendReply(CredentialManager.Instance.GetAllCredentials());
                    }
                    else if (CurrentAction is ListPeers)
                    {
                        SendReply(PeerManager.Instance.GetAllBroadcastObjects());
                    }
                    else if (CurrentAction is ListObjects)
                    {
                        PhiServer.Instance.ObjectManagerChannel.Send(new InternalListObjects(fromObjectManager));
                        SendReply(fromObjectManager.Receive());//consider splitting
                    }
                    else if (CurrentAction is WhoIAm)
                    {
                        IPAddress newIP = (CurrentAction as WhoIAm).IP;
                        if (ExternalPeer != null)
                        {
                            BroadcastObject newBC = new BroadcastObject(ExternalPeer.BC.PublicKey, newIP, ExternalPeer.BC.Port);
                            PeerManager.Instance.Change(ExternalPeer.BC, newBC);
                        }
                        SendReply(ReturnCode.Success);
                    }
                    else if (CurrentAction is WhoAmI)
                    {
                        SendReply(((IPEndPoint)connection.Client.RemoteEndPoint).Address);//send remote IP back
                    }
                    else if (CurrentAction is GetObjectInformation)
                    {
                        PhiServer.Instance.ObjectManagerChannel.Send(new InternalGetStats((CurrentAction as GetObjectInformation).Key, fromObjectManager));
                        SendReply(fromObjectManager.Receive());
                    }
                    else if (CurrentAction is ObjectCopy)
                    {
                        CreateCopy((ObjectCopy)CurrentAction);
                    }
                    else if (CurrentAction is UpdateCopy)
                    {
                        UpdateCopy((UpdateCopy)CurrentAction);
                    }
                    else if (CurrentAction is ListDeleteCopy)
                    {
                        DeleteEntriesForwarded((ListDeleteCopy)CurrentAction);
                    }
                    else if (CurrentAction is GetData)
                    {
                        GetData((GetData)CurrentAction);
                    }
                    else if (CurrentAction is DeleteCopy)
                    {
                        DeleteCopy((DeleteCopy)CurrentAction);
                    }
                    else if (CurrentAction is Gossip)
                    {
                        SendReply(ReturnCode.Success); //yay we are still connected!
                        PeerManager.Instance.RemovePeer(null); //check all connections
                    }
                    else if (CurrentAction is UpdateServerCopy)
                    {
                        UpdateCopyOnPeer((UpdateServerCopy)CurrentAction);
                    }
                    else
                    {
                        Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), CurrentAction.ToString() + " is an unknown package");
                        SendReply(ReturnCode.Failure);
                    }
                }
                else
                {
                    Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), CurrentAction.ToString() + " is an unknown client package");
                    SendReply(ReturnCode.Failure);
                }

            }

        }

        private void UpdateCopyOnPeer(UpdateServerCopy package)
        {
            SendReply(ReturnCode.Success);
            currentObjectName = package.Key;
            ReturnCode result = LockEntry();
            if (result == ReturnCode.Success)
            {
                if (objectLock.Coordinator && 
                    !objectLock.ReplicatingPeers.Contains(this.externalPeer) && //not already replicating
                    package.ObjectInfo.Version != objectLock.ObjectInformation.Version) //there is a difference
                {
                    objectLock.ReplicatingPeers.Add(this.externalPeer);

                    Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), externalPeer.BC.ToString() + " is getting a fresh copy of " + package.Key);

                    result = CreateNewCopy(externalPeer); //could be optimized for !fastreturn

                    if (objectLock.Policy.EnforceAtomicity)
                    {
                        PhiPackage abortCommit;
                        if (result == ReturnCode.Success)
                        {
                            abortCommit = new Commit(package.Key);

                        }
                        else
                        {
                            abortCommit = new Abort(package.Key);
                        }
                        ForwardedPackage fa = new ForwardedPackage(abortCommit, null);
                        try
                        {
                            externalPeer.PeerWorkerChannel.Send(fa);
                        }
                        catch (PoisonException)
                        {
                            //connection lost
                        }
                    }

                    Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), externalPeer.BC.ToString() + " got a fresh copy of " + package.Key);
                }
                ReleaseLockEntry();
            }
        }

        private void Abort(Abort abort)
        {
            
            doAbort(abort.Key);

            SendReply(ReturnCode.Success);
        }

        private void doAbort(string key)
        {
            currentObjectName = key;
            objectLock = locks.FetchLock(key);
            ObjectLock oe = objectLock;

            oe.Device.Abort(key); //ABORT

            //read older version of object information and load into ObjectEntry
            oe.ObjectInformation = oe.Device.ReadObjectInformation(key);
            ReleaseLockEntry();
        }

        private void Commit(Commit commit)
        {
            currentObjectName = commit.Key;
            objectLock = locks.FetchLock(commit.Key);
            ObjectLock oe = objectLock;

            oe.Device.Commit(commit.Key); //COMMIT
            ReturnCode result = ReturnCode.Success;

            if (!oe.Device.Exist(commit.Key)) //object was deleted
            {
                PhiServer.Instance.ObjectManagerChannel.Send(new RemoveObject(commit.Key, fromObjectManager));
                result |= (ReturnCode)fromObjectManager.Receive();
            }
            ReleaseLockEntry();

            SendReply(result);
        }

        private void AddRemotePeer(AddRemotePeer package) //forwarded peer!
        {
            if (PeerManager.Instance.Exist(package.IP, package.Port))
            {
                SendReply(ReturnCode.Success); //already connected
            }
            else
            {
                PeerWorker pc = new PeerWorker();
                ReturnCode res = pc.ConnectAndLaunch(package.IP, package.Port);
                if (res == ReturnCode.Success)
                {
                    if (PeerManager.Instance.AddPeer(pc.Peer))
                    {
                        try
                        {
                            pc.Peer.PeerWorkerChannel.Send(
                                      new ForwardedPackage(new AddPeer(PhiServer.Instance.BroadcastObject.IP, PhiServer.Instance.BroadcastObject.Port)
                                                         , null)
                                                         );
                        }
                        catch (PoisonException)
                        {
                            //crap - connection to the peer isn't available
                        }
                    }
                    else
                    {
                        pc.Peer.PeerWorkerChannel.Poison();
                    }
                }
                SendReply(res);
            }
        }

        private void AddPeer(AddPeer package)
        {
            IList<PeerContainer> peers = PeerManager.Instance.GetPeers();

            PeerWorker pc = new PeerWorker();
            IPEndPoint remoteIp = (IPEndPoint)connection.Client.RemoteEndPoint;

            //HACK - IP may be loopback (localhost)
            IPAddress toBroadcast = IPAddress.IsLoopback((IPAddress)remoteIp.Address) ? package.IP : remoteIp.Address;

            if (PeerManager.Instance.Exist(toBroadcast, package.Port))
            {
                ExternalPeer = PeerManager.Instance.GetPeer(new BroadcastObject(toBroadcast, package.Port));
                SendReply(ReturnCode.Success); //already connected
            }
            else
            {
                ReturnCode res = pc.ConnectAndLaunch(toBroadcast, package.Port);


                SendReply(res);

                if (res == ReturnCode.Success && PeerManager.Instance.AddPeer(pc.Peer))
                {
                    ExternalPeer = pc.Peer;
                    try
                    {
                        pc.Peer.PeerWorkerChannel.Send(
                                   new ForwardedPackage(new AddPeer(PhiServer.Instance.BroadcastObject.IP, PhiServer.Instance.BroadcastObject.Port)
                                                      , null)
                                                      );
                    }
                    catch (PoisonException)
                    {
                        //oooh crap!
                        throw new WebException("Unable to mark self @ peer");
                    }

                }
                else
                {
                    ExternalPeer = PeerManager.Instance.GetPeer(toBroadcast, package.Port);

                    if (pc.Peer != null && pc.Peer.PeerWorkerChannel != null)
                        pc.Peer.PeerWorkerChannel.Poison();
                }

                if (res == ReturnCode.Success)
                {//retransmit to all peers!

                    foreach (PeerContainer p in peers)
                    {
                        //this may fail if IP is within a LAN
                        try
                        {
                            ForwardedPackage fwd = new ForwardedPackage(new AddRemotePeer(toBroadcast, package.Port), null);
                            p.PeerWorkerChannel.Send(fwd);
                        }
                        catch (PoisonException)
                        {
                            //peer no longer online
                        }

                    }
                }
            }
        }

        #region helpers

        private ReturnCode LockEntry()
        {
            LockObjectEntry lockAction = new LockObjectEntry(currentObjectName, fromObjectManager);
            PhiServer.Instance.ObjectManagerChannel.Send(lockAction);

            //reply is either ObjectEntry / ReturnCode
            object reply;

        keepaliveloop:
            try
            {
                reply = fromObjectManager.Receive(1000);
            }
            catch (TimeoutException)
            {
                SendReply(ReturnCode.WorkingOnObject);
                goto keepaliveloop;
            }

            if (reply is ReturnCode)
            {
                return (ReturnCode)reply;
            }
            else
            {
                objectLock = (ObjectLock)reply;
                return ReturnCode.Success;
            }
        }
        private void ReleaseLockEntry()
        {
            PhiServer.Instance.ObjectManagerChannel.Send(new UnlockObjectEntry(currentObjectName));
            objectLock = null;
        }

        private void SendReply(object reply)
        {
            object toSend = reply;
            if (aes != null)
                toSend = Encryption.Encrypt(toSend, aes, formatter);

            formatter.Serialize(outputStream, toSend);
            outputStream.Flush();
        }

        private PhiPackage ReceiveAction()
        {
            PhiPackage result = (PhiPackage)formatter.Deserialize(inputStream);

            if (result is RSAEncrypted)
                result = (PhiPackage)Encryption.RSADecrypt((RSAEncrypted)result, rsa, formatter);


            if (aes != null && result is Encrypted)
                result = (PhiPackage)Encryption.Decrypt((Encrypted)result, aes, formatter);


            return result;
        }

        public bool isConnectionLost(object o)
        {
            return (o is ReturnCode) && ((ReturnCode)o == ReturnCode.ConnectionLost);
        }
        #endregion
    }
}
